package cn.doitedu.cn.doitedu.rtdw.etl;

import cn.doitedu.cn.doitedu.rtdw.ddl.CatalogDatabaseSwitch;
import cn.doitedu.cn.doitedu.rtdw.udfs.TimestampTrunc;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job4_VideoPlayOlapSupport {

    public static void main(String[] args) {


        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 切换catalog为hiveCatalog
        CatalogDatabaseSwitch.switchCatalogDatabase("doit40-hive","doit40_rtdw",tenv);

        // 注册函数
        tenv.createTemporaryFunction("time_trunc", TimestampTrunc.class);

        // 创建doris目标表的，flinksql映射表
        tenv.executeSql("drop table if exists video_events_agg01_doris");
        tenv.executeSql(
                " CREATE TABLE video_events_agg01_doris(            "+
                        "          dt                           DATE         "+
                        "         ,time_60                    string         "+
                        "         ,time_05                    string         "+
                        "         ,user_id                    bigint         "+
                        "         ,member_level               int            "+
                        "         ,device_type                string         "+
                        "         ,province                   string         "+
                        "         ,page_url                   string         "+
                        "         ,video_id                   bigint         "+
                        "         ,play_tracking_id           string         "+
                        "         ,video_album                string         "+
                        "         ,video_time_amt             bigint         "+
                        "         ,play_time_amt              bigint         "+
                        " ) WITH (                                           "+
                        "    'connector' = 'doris',                          "+
                        "    'fenodes' = 'doitedu:8030',                     "+
                        "    'table.identifier' = 'dws.video_events_agg01',  "+
                        "    'username' = 'root',                            "+
                        "    'password' = 'root',                            "+
                        "    'sink.label-prefix' = 'doris_label-oop'         "+
                        " )                                                  "
        );




        // 在流模式下，sql中的over窗口，要求排序列必须是一个时间属性列
        tenv.executeSql(
                "create temporary view aggregated_view AS  " +
                        "with tmp1 as (                                                                          \n "+
                        " select                                                                                 \n "+
                        "   user_id,                                                                             \n "+
                        "   event_id,                                                                            \n "+
                        "   device_type,                                                                         \n "+
                        "   page_url,                                                                            \n "+
                        "   member_level_id,                                                                     \n "+
                        "   gps_province as province,                                                            \n "+
                        "   cast( properties['video_id'] as int)  as video_id,                                   \n "+
                        "   properties['play_id'] as play_id,                                                    \n "+
                        "   event_time,                                                                          \n "+
                        "   if(event_id = 'video_pause',1,0) as flag,                                            \n "+
                        "   rt                                                                                   \n "+
                        " from dwd_kafka                                                                         \n "+
                        " where event_id in ('video_play','video_pause','video_resume','video_stop','video_hb')  \n "+
                        " )                                                                                      \n "+
                        " ,tmp2 as (                                                                             \n "+
                        " select                                                                                 \n "+
                        "   user_id,                                                                             \n "+
                        " 	video_id,                                                                            \n "+
                        " 	event_id,                                                                            \n "+
                        " 	play_id,                                                                             \n "+
                        " 	event_time,                                                                          \n "+
                        " 	device_type,                                                                         \n "+
                        " 	province,                                                                            \n "+
                        " 	page_url,                                                                            \n "+
                        " 	member_level_id,                                                                     \n "+
                        " 	sum(flag) over(partition by play_id order by rt) - flag as flag,                     \n "+
                        " 	rt                                                                                   \n "+
                        " from tmp1                                                                              \n "+
                        " )                                                                                      \n "+
                        "                                                                                        \n "+
                        " select                                                                                 \n "+
                        "   to_date(date_format(window_end,'yyyy-MM-dd')) as dt,                                 \n "+
                        " 	time_trunc(cast(window_end as string),60) as time_60,                                \n "+
                        " 	time_trunc(cast(window_end as string),5) as time_05,                                 \n "+
                        " 	user_id,                                                                             \n "+
                        " 	member_level_id,                                                                     \n "+
                        " 	device_type,                                                                         \n "+
                        " 	province,                                                                            \n "+
                        " 	page_url,                                                                            \n "+
                        " 	video_id,                                                                            \n "+
                        " 	play_id as play_tracking_id,                                                         \n "+
                        " 	proctime() as pt,                                                                    \n "+
                        " 	max(event_time) - min(event_time) as play_time_long                                  \n "+
                        " from table(                                                                            \n "+
                        "    tumble(table tmp2,descriptor(rt),interval '5' minute)                               \n "+
                        " )                                                                                      \n "+
                        " group by                                                                               \n "+
                        "   window_start,                                                                        \n "+
                        " 	window_end,                                                                          \n  "+
                        " 	user_id,                                                                             \n  "+
                        " 	member_level_id,                                                                     \n  "+
                        " 	device_type,                                                                         \n  "+
                        " 	province,                                                                            \n  "+
                        " 	page_url,                                                                            \n  "+
                        " 	video_id,                                                                            \n  "+
                        " 	play_id,                                                                             \n  "+
                        " 	flag                                                                                 \n  "
        );

        // 关联视频的维度信息：视频专辑，视频时长
        tenv.executeSql(
                "insert into  video_events_agg01_doris " +
                "SELECT                                      "+
                "  a.dt                                      "+
                " ,a.time_60                                 "+
                " ,a.time_05                                 "+
                " ,a.user_id                                 "+
                " ,a.member_level_id as member_level         "+
                " ,a.device_type                             "+
                " ,a.province                                "+
                " ,a.page_url                                "+
                " ,a.video_id                                "+
                " ,a.play_tracking_id                        "+
                " ,b.video_album                             "+
                " ,b.video_timelong as video_time_amt        "+
                " ,a.play_time_long as play_time_amt         "+
                " from aggregated_view as a                  "+
                " left join dim_video_hbase                  "+
                " for system_time as of a.pt as b            "+
                " on a.video_id = b.id                       ");




    }

}
